package com.yl.util;

import cn.hutool.core.util.StrUtil;
import com.yl.constant.Const;
import org.apache.flink.api.java.utils.ParameterTool;

import java.io.IOException;

/**
 * @author wlf
 * @apiNote 带环境的参数配置读取工具类
 * @since 2022/8/5
 */
public class ParamUtil {

    /**
     * 按照优先级获取参数配置读取工具类
     */
    public static ParameterTool getParameterTool(String[] args) throws IOException {
        // 系统环境变量参数 -Dkey=value
        ParameterTool sysConfig = ParameterTool.fromSystemProperties();
        // 启动 main 参数 -key value 或者 --key value
        ParameterTool argsConfig = ParameterTool.fromArgs(args);
        // 默认配置文件参数 app.properties
        ParameterTool defaultConfig = ParameterTool
                .fromPropertiesFile(Thread.currentThread().getContextClassLoader().getResourceAsStream(Const.CFG_DEFAULT));
        // 按照优先级获取有效环境值
        String activeEnv = getActiveEnv(sysConfig, argsConfig, defaultConfig);
        String envConfig = StrUtil.format(Const.CFG_ENV_TPL, activeEnv);
        // 读取执行环境配置文件参数
        ParameterTool currentEnvConfig = ParameterTool
                .fromPropertiesFile(Thread.currentThread().getContextClassLoader().getResourceAsStream(envConfig));
        // 参数优先级 系统参数 > 启动参数 > 执行环境参数app-{env}.properties > 默认参数app.properties
        return defaultConfig
                .mergeWith(currentEnvConfig)
                .mergeWith(argsConfig)
                .mergeWith(sysConfig);
    }

    /**
     * 按照优先级获取有效环境值
     */
    private static String getActiveEnv(ParameterTool sysConfig, ParameterTool argsConfig, ParameterTool defaultConfig) {
        //选择参数环境
        String env;
        if (sysConfig.has(Const.APP_ENV)) {
            env = sysConfig.get(Const.APP_ENV);
        } else if (argsConfig.has(Const.APP_ENV)) {
            env = argsConfig.get(Const.APP_ENV);
        } else if (defaultConfig.has(Const.APP_ENV)) {
            env = defaultConfig.get(Const.APP_ENV);
        } else {
            //如果没有配置抛出异常
            throw new IllegalArgumentException("Please set up the environment.");
        }
        return env;
    }

    /**
     * 例如：kafka.bootstrap.servers -> bootstrap.servers
     *
     * @param paramKey 参数 key
     * @return 去掉前缀的参数 key
     */
    public static String getKeyWithoutPrefix(String paramKey) {
        return paramKey.substring(paramKey.indexOf(Const.PTN_PERIOD) + 1);
    }

}
